#pragma once

#include "muduo/net/TcpConnection.h"
#include "muduo/protobuf/codec.h"

#include "../mqcommon/mq_logger.hpp"
#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include "../mqcommon/mq_proto.pb.h"

#include "mq_consumer.hpp"
#include <mutex>
#include <condition_variable>
#include <unordered_map>

namespace mq
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using basicCommonResponsePtr = std::shared_ptr<basicCommonResponse>;
    using basicConsumerResponsePtr = std::shared_ptr<basicConsumerResponse>;
    class Channel
    {
    private:
        std::string _cid;
        muduo::net::TcpConnectionPtr _conn;
        ProtobufCodecPtr _codec; 
        Consumer::ptr _consumer;

        //由于muduo库的发送和接收都是异步的，例如我们声明一个交换机，这个请求可能还在发送缓冲区，并没有发送，我们此时如果去给这个交换机推送消息就会出问题。因此需要我们⾃⼰在收到响应后，通过判断是否是等待的指定响应来进⾏同步
        std::mutex _mutex;
        std::condition_variable _cv;
        std::unordered_map<std::string,basicCommonResponsePtr> _basic_resp;
    private:
        basicCommonResponsePtr waitResponse(std::string &rid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cv.wait(lock,[&rid,this](){
                return _basic_resp.find(rid) != _basic_resp.end();
            });

            basicCommonResponsePtr basic_resp = _basic_resp[rid];
            _basic_resp.erase(rid);
            return basic_resp;
        }
    public:
    using ptr = std::shared_ptr<Channel>;
        //cid是我们随机生成的,不需要外部传入
        Channel(const muduo::net::TcpConnectionPtr& conn,const ProtobufCodecPtr & codec):
            _conn(conn),_codec(codec)
        {
            _cid = UUIDHelper::uuid();
            DLOG("new channel :%p",this);
        }
        //在信道关闭时，给服务器发送一个取消订阅请求
        ~Channel(){
            DLOG("delete channel :%p",this);
            basicCancel();
        }
        std::string cid(){ return _cid; }
    public:
        //这两个接口是向服务端发送请求，在服务端创建对应信道
        bool openChannel(){
            //组织请求
            openChannelRequest req;
            std::string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);

            //发送
            _codec->send(_conn,req);
            basicCommonResponsePtr resp = waitResponse(rid);
            return resp->ok();

        }
        void closeChannel(){
            //组织请求
            closeChannelRequest req;
            std::string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);

            //发送
            _codec->send(_conn,req);
            waitResponse(rid);
            return;
        }
        
        bool declareExchange(const std::string &exchange_name,
                            ExchangeType type, bool durable, bool auto_delete,
                            google::protobuf::Map<std::string, std::string> &args){
            declareExchangeRequest req;
            std::string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(exchange_name);
            req.set_exchange_type(type);
            req.set_durable(durable);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);

            //然后向服务器发送请求
            _codec->send(_conn,req);
            //等待服务器的响应
            basicCommonResponsePtr resp = waitResponse(rid);
            //返回
            return resp->ok();
        }
        void deleteExchange(const std::string &exchange_name){
            deleteExchangeRequest req;
            std::string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(exchange_name);

            _codec->send(_conn,req);
            waitResponse(rid);
            return;
        }

        bool declareQueue(const std::string &queue_name,
                        bool exclusive,bool durable, bool auto_delete,
                        google::protobuf::Map<std::string, std::string> &args){
            
            declareQueueRequest req;
            std::string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(queue_name);
            req.set_exclusive(exclusive);
            req.set_durable(durable);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);

            //然后向服务器发送请求
            _codec->send(_conn,req);
            //等待服务器的响应
            basicCommonResponsePtr resp = waitResponse(rid);
            //返回
            return resp->ok();
        }
        void deleteQueue(const std::string &queue_name)
        {
            deleteQueueRequest req;
            std::string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(queue_name);

            _codec->send(_conn,req);
            waitResponse(rid);
            return;
        }

        bool queueBind(const std::string &exchange_name,const std::string &queue_name,const std::string &binding_key)
        {
            queueBindRequest req;
            std::string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(exchange_name);
            req.set_queue_name(queue_name);
            req.set_binding_key(binding_key);
            
            //然后向服务器发送请求
            _codec->send(_conn,req);
            //等待服务器的响应
            basicCommonResponsePtr resp = waitResponse(rid);
            //返回
            return resp->ok();
        }
        void queueUnBind(const std::string &exchange_name,const std::string &queue_name){
            queueUnBindRequest req;
            std::string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(exchange_name);
            req.set_queue_name(queue_name);

            _codec->send(_conn,req);
            waitResponse(rid);
            return;
        }

        void basicPublish(const std::string &exchange_name,BasicProperties *bp,const std::string &body){
            basicPublishRequest req;
            std::string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(exchange_name);
            req.set_body(body);
            if(bp != nullptr)
            {
                req.mutable_properties()->set_id(bp->id());
                req.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                req.mutable_properties()->set_routing_key(bp->routing_key());
            }

            //然后向服务器发送请求
            _codec->send(_conn,req);
            //等待服务器的响应
            waitResponse(rid);
            return;

        }
        void basicAck(const std::string &msg_id)
        {
            basicAckRequest req;
            std::string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(_consumer->_qname);
            req.set_message_id(msg_id);

            //然后向服务器发送请求
            _codec->send(_conn,req);
            //等待服务器的响应
            waitResponse(rid);
            return;
        }

        bool basicConsume(const std::string &consume_tag,const std::string &qname,bool auto_ack,const ConsumerCallBack &cb)
        {
            if(_consumer.get() != nullptr)
            {
                DLOG("当前信道已订阅其他队列消息！");
                return false;
            }

            basicConsumeRequest req;
            std::string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_consumer_tag(consume_tag);
            req.set_queue_name(qname);
            req.set_auto_ack(auto_ack);

            //然后向服务器发送请求
            _codec->send(_conn,req);
            //等待服务器的响应
            basicCommonResponsePtr resp = waitResponse(rid);
            if (resp->ok() == false) {
                DLOG("添加订阅失败！");
                return false;
            }

            _consumer = std::make_shared<Consumer>(qname,consume_tag,auto_ack,cb);
            return true;
        }
        void basicCancel(){
            if(_consumer.get() == nullptr)
            {
                return ;
            }

            basicCancelRequest req;
            std::string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_consumer_tag(_consumer->_ctag);
            req.set_queue_name(_consumer->_qname);

            //然后向服务器发送请求
            _codec->send(_conn,req);
            //等待服务器的响应
            waitResponse(rid);
            //将客户端的消费者清除，置为null
            _consumer.reset();
            return ;
        }

    public:
    //这两个函数是给Connection提供的

        //连接收到基础响应后，向hash_map中添加响应
        void putBasicResponse(const basicCommonResponsePtr& resp) {
            std::unique_lock<std::mutex> lock(_mutex);
            _basic_resp.insert(std::make_pair(resp->rid(),resp));
            _cv.notify_all();
        }

        //连接收到消息推送后，需要通过信道找到对应的消费者对象，通过回调函数进⾏消息处理
        void consume(const basicConsumerResponsePtr& resp) {
            if (_consumer.get() == nullptr) {
                DLOG("消息处理时，未找到订阅者信息！");
                return;
            }
            if (_consumer->_ctag != resp->consumer_tag()) {
                DLOG("收到的推送消息中的消费者标识，与当前信道消费者标识不⼀致！");
                return ;
            }

            _consumer->_cb(resp->consumer_tag(),resp->mutable_properties(),resp->body());
            return ;
        }
    };

    class ChannelManager
    {
    private:
        std::mutex _mutex;
        std::unordered_map<std::string,Channel::ptr> _channels;
    public:
    using ptr = std::shared_ptr<ChannelManager>;
        ChannelManager(){}
        Channel::ptr create(const muduo::net::TcpConnectionPtr& conn,const ProtobufCodecPtr & codec)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto channel = std::make_shared<Channel>(conn, codec);
            _channels.insert(std::make_pair(channel->cid(), channel));
            return channel;

        }
        void remove(const std::string &cid){
            std::unique_lock<std::mutex> lock(_mutex);
            _channels.erase(cid);
        }
        Channel::ptr get(const std::string &cid){
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(cid);
            if(it == _channels.end())
            {
                return Channel::ptr();
            }

            return it->second;
        }
    };
};